package etcd

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"go.etcd.io/etcd/clientv3"
	"time"
)

var LastEntries []*LogEntry

/*
新建 etcd 连接

参数:
	endpoints []string,
	timeout time.Duration

返回:
	cli *clientv3.Client,
	err error
*/
func NewClient(endpoints []string, timeout time.Duration) (cli *clientv3.Client, err error) {
	cli, err = clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: timeout,
	})
	if err != nil {
		fmt.Println("[pkg etcd] connect to etcd failed,err:", err)
		return nil, err
	}
	fmt.Println("[pkg etcd] connect etcd success")
	return
}

// 日志入口配置
type LogEntry struct {
	Path  string `json:"path"`
	Topic string `json:"topic"`
}

func getLogEntry(originValue []byte) ([]*LogEntry, error) {
	var value []*LogEntry
	err := json.Unmarshal(originValue, &value)
	if err != nil {
		return nil, err
	}
	return value, nil
}

/*
从etcd中获取配置信息

参数：
	cli *clientv3.Client,
	key string
返回：
	[]*LogEntry,
	error
*/
// 根据 key 获得 value(这个value是个配置)
//
// 初始化时使用，使用了 etcd 的 watch 功能后将不再使用这个函数
func GetConf(cli *clientv3.Client, key string) ([]*LogEntry, error) {
	var value []*LogEntry
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	resp, err := cli.Get(ctx, key)
	cancel()
	if err != nil {
		fmt.Println("[pkg etcd] get key err:", err)
		return nil, err
	}
	//fmt.Printf("%#v\n", resp)
	for _, ev := range resp.Kvs {
		//fmt.Printf("%s:%s\n", ev.Key, ev.Value)
		value, err = getLogEntry(ev.Value)
		if err != nil {
			fmt.Println("[pkg etcd] getLogEntry err:", err)
			return nil, err
		}
	}
	LastEntries = value
	return value, nil
}

//测试用
func PutValue(cli *clientv3.Client, key string) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	_, err := cli.Put(ctx, key, `[{"path":"/tmp/nginx.log","topic":"web-log"},{"path":"/tmp/redis.log","topic":"web-log"}]`)
	cancel()
	if err != nil {
		fmt.Println(err)
		return err
	}
	return nil
}

//测试用
func PutValue2(cli *clientv3.Client, key string) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	_, err := cli.Put(ctx, key, `[{"path":"/tmp/nginx.log","topic":"web-log"},{"path":"/tmp/redis.log","topic":"web-log"},{"path":"/tmp/go.log","topic":"web-log"}]`)
	cancel()
	if err != nil {
		fmt.Println(err)
		return err
	}
	return nil
}

// 使用 etcd 的监听功能，监听某个key，并对key对应的新值进行 Diff 操作。
//
// 将 Diff 函数返回的 minus 和 plus 传递给对调函数 serveFunc
//
// 由于 serveFunc 可能需要使用 kafka，所以这里需要传入 producer sarama.SyncProducer
func WatchAndServe(cli *clientv3.Client, key string, producer sarama.SyncProducer, serveFunc func(eventType string, minus map[int]*LogEntry, plus map[int]*LogEntry, producer sarama.SyncProducer)) {
	wChan := cli.Watch(context.Background(), key)
	for wResp := range wChan {
		for _, ev := range wResp.Events {
			fmt.Printf("[pkg etcd] Type: %s Key: %s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
			value, err := getLogEntry(ev.Kv.Value)
			if err != nil {
				fmt.Println("[pkg etcd] getLogEntry err:", err)
			}
			minus, plus := Diff(value)
			// 更新 LastEntries
			LastEntries = value
			// serveFunc 需要参数 eventType, minus, plus, producer
			go serveFunc(ev.Type.String(), minus, plus, producer)
		}
	}
}

// 对新的ev.Kv.Value 和上一个 ev.Kv.Value 进行 Diff，返回减少的值(minus)和增加的值(plus)。
// 这个函数将会在 WatchAndServe 函数中调用
func Diff(value []*LogEntry) (map[int]*LogEntry, map[int]*LogEntry) {
	var minus = make(map[int]*LogEntry)
	var plus = make(map[int]*LogEntry)
	var common = make(map[string]int)
	var common2 = make(map[string]int)
	for k, v := range LastEntries {
		for k2, v2 := range value {
			if v2.Path == v.Path && v2.Topic == v.Topic {
				common[v.Path] = k
				common2[v2.Path] = k2
			}
		}
	}
LOOP:
	for k, v := range value {
		for k2, v2 := range common2 {
			fmt.Println(k, k2, v2)
			if k == v2 {
				continue LOOP
			}
		}
		plus[k] = v
	}
LOOP2:
	for k, v := range LastEntries {
		for k2, v2 := range common {
			fmt.Println(k, k2, v2)
			if k == v2 {
				continue LOOP2
			}
		}
		minus[k] = v
	}
	return minus, plus
}
